package com.bcxj.Test;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.Date;
import java.util.Random;

public class OrderStatistics {

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 模拟订单数据源，运行10分钟后结束
        DataStream<Order> orderStream = env.addSource(new SourceFunction<Order>() {
            private volatile boolean isRunning = true;
            private Random random = new Random();
            private long endTime = System.currentTimeMillis() + 10 * 1000; // 10s后结束

            @Override
            public void run(SourceContext<Order> ctx) throws Exception {
                int count = 0;
                while (isRunning && System.currentTimeMillis() < endTime) {
                    long timestamp = System.currentTimeMillis();
                    String orderId = "order_" + random.nextInt(10000);
                    double amount = random.nextDouble() * 100;
                    if (count > 50) {
                        timestamp = System.currentTimeMillis() + 60 * 60 * 1000 * 2;
                    }
                    Order order = new Order(orderId, timestamp, amount);
                    ctx.collect(order);
                    System.out.println(order + ": " + count);
                    Thread.sleep(100); // 每100ms生成一个订单
                    count++;
                }
                cancel(); // 超过结束时间后调用cancel方法
            }

            @Override
            public void cancel() {
                isRunning = false;
            }
        });

        // 提取时间戳和生成水印
        SingleOutputStreamOperator<Order> timestampedOrderStream = orderStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Order>forMonotonousTimestamps()
                                .withTimestampAssigner((event, timestamp) -> event.timestamp)
                );

        // 定义窗口操作，每小时统计一次
        timestampedOrderStream
                .keyBy(order -> 1) // 针对所有订单进行全局统计
                .window(TumblingEventTimeWindows.of(Time.minutes(10)))
                .apply((Integer key, TimeWindow window, Iterable<Order> input, Collector<String> out) -> {
                    int count = 0;
                    double totalAmount = 0.0;
                    for (Order order : input) {
                        count++;
                        totalAmount += order.amount;
                    }
                    Date start = new Date(window.getStart());
                    Date end = new Date(window.getEnd());
                    out.collect("Window: " + "{" + start + " :: " + end + "}" + ", Order Count: " + count + ", Total Amount: " + totalAmount);
                }).returns(String.class)
                .print();

        env.execute("Order Statistics Job");
    }
}
